package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo12GroupByKey {
  def main(args: Array[String]): Unit = {

    /**
     * groupByKey：转换算子，需要作用在KV格式的RDD上，会按照K进行分组
     * 最终相同K的所有的V会分到同一组
     * 可以进行任意的聚合操作
     */

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName(this.getClass.getSimpleName.replace("$", ""))
    val sc = new SparkContext(conf)

    val wordsRDD: RDD[String] = sc.parallelize(List("a", "a", "a", "b", "b", "b"))

    wordsRDD
      // 无法直接调用groupByKey算子，因为RDD是单Value的格式，并不是KV格式
      .groupBy(w => w)
      .map(kv => s"${kv._1},${kv._2.size}")
      .foreach(println)

    wordsRDD
      .map(w => (w, 1)) // 将每个单词变成KV格式，以单词本身作为K，1作为V
      .groupByKey() // 自动的对Key进行分组
      .map(kv => s"${kv._1},${kv._2.size}")
      .foreach(println)


  }

}
